package cn.kimming.rpc.remoting.client;

import cn.kimming.rpc.common.exception.RpcException;
import cn.kimming.rpc.common.util.ClassUtil;
import cn.kimming.rpc.remoting.codec.MyRpcProtocolClientCodec;
import cn.kimming.rpc.common.constant.LoadBalance;
import cn.kimming.rpc.remoting.registry.InstanceDetail;
import cn.kimming.rpc.remoting.registry.ClientRpcRegistry;
import cn.kimming.rpc.remoting.strategy.LeastCallStrategy;
import cn.kimming.rpc.remoting.transport.RpcRequest;
import cn.kimming.rpc.remoting.transport.RpcResponse;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.ReadTimeoutHandler;
import org.apache.curator.x.discovery.ProviderStrategy;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.ServiceProvider;
import org.apache.curator.x.discovery.strategies.RandomStrategy;
import org.apache.curator.x.discovery.strategies.RoundRobinStrategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

public class RpcClient {
    private static final Logger logger = LoggerFactory.getLogger(RpcClient.class);
    private String serviceName;
    private RpcClientConfig config;
    private ServiceProvider<InstanceDetail> provider;
    ServiceDiscovery<InstanceDetail> discovery;
    private Bootstrap bootstrap;
    private NioEventLoopGroup group;

    public RpcClient(String serviceName, RpcClientConfig config) {
        this.serviceName = serviceName;
        this.config = config;
        init();
    }

    private void init() {
        logger.info("[my-rpc] init client {}", toString());
        this.discovery = ClientRpcRegistry.getDiscovery(config.getRegistryUrl());
        this.provider = initProvider();
        this.group = new NioEventLoopGroup();
        this.bootstrap = new Bootstrap()
                .group(this.group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline()
                                .addLast(new MyRpcProtocolClientCodec(config.getSerializationWay()))
                                .addLast(new ReadTimeoutHandler(config.getTimeout(), TimeUnit.MILLISECONDS))
                                .addLast(new ClientHandler());
                    }
                });
    }

    private ServiceProvider<InstanceDetail> initProvider() {
        ProviderStrategy<InstanceDetail> strategy = null;

        LoadBalance loadBalance = config.getLoadBalance();
        switch (loadBalance) {
            case ROUNDROBIN:
                strategy = new RoundRobinStrategy<>();
                break;
            case RANDOM:
                strategy = new RandomStrategy<>();
                break;
            case LEASTCALL:
                strategy = new LeastCallStrategy();
                break;
        }

        ServiceProvider<InstanceDetail> provider = discovery.serviceProviderBuilder().serviceName(serviceName).providerStrategy(strategy).build();
        try {
            provider.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return provider;
    }

    public RpcResponse doRpc(Method method, Object[] args) throws RpcException {

        Object[] ipAndPort = chooseRoute();
        String ip = String.valueOf(ipAndPort[0]);
        int port = (int) ipAndPort[1];

        logger.info("[my-rpc] client try to connect {}:{}", ip, port);
        ChannelFuture channelFuture = null;
        try {
            channelFuture = bootstrap.connect(ip, port).sync();
        } catch (InterruptedException e) {
            logger.error("[my-rpc] client try to connect "+ ip +":" + port + " failed", e);
            e.printStackTrace();
        }

        Channel channel = channelFuture.channel();
        ClientHandler clientHandler = channel.pipeline().get(ClientHandler.class);

        RpcRequest request = buildRequest(method, args);

        InetSocketAddress inetSocketAddress = (InetSocketAddress) channel.remoteAddress();
        logger.info("[my-rpc] client 请求主机{}:{}上的服务{}, requestId: {}", inetSocketAddress.getHostString(), inetSocketAddress.getPort(), request.getInterfaceName() + ":" + request.getMethodName(), request.getRequestId());
        channel.writeAndFlush(request).addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
        try {
            channel.closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return clientHandler.getResponse();
    }

    /**
     * 选择ip和端口
     * @return
     */
    private Object[] chooseRoute() {
        if (!config.useRegistry()) {
            return new Object[]{config.getIp(), config.getPort()};
        }

        // 如果没有指定主机或者指定值不合法, 则从注册中心获取
        try {
            ServiceInstance<InstanceDetail> instance = provider.getInstance();
            String address = instance.getAddress();
            Integer port = instance.getPort();
            return new Object[]{address, port};
        } catch (Exception e) {
            e.printStackTrace();
        }
        throw new RpcException(serviceName + "没有可用的服务");
    }


    private RpcRequest buildRequest(Method method, Object[] args) {
        RpcRequest request = new RpcRequest();
        request.setRequestId(UUID.randomUUID().toString());
        request.setInterfaceName(serviceName);
        request.setMethodName(method.getName());
        String[] methodTypeNames = ClassUtil.parseClasses2Names(method.getParameterTypes());
        request.setMethodTypeNames(methodTypeNames);
        request.setMethodArgs(args);
        return request;
    }

    @Override
    public String toString() {
        return "RpcClient{" +
                "serviceName='" + serviceName + '\'' +
                ", config=" + config +
                '}';
    }

}
